package breadbox.dataManagement;


import java.util.SortedSet;
import java.util.Timer;
import java.util.TimerTask;
import java.util.TreeSet;

import breadbox.pastry.ID;
import breadbox.pastry.LongID;
import breadbox.pastry.Message;
import breadbox.pastry.routing.LeafSet;
import breadbox.pastry.routing.RoutingNode;

public class NetworkedReplicationManager extends ReplicationManager {
	//node that is 'closest' to chunkID is in charge of replication
	//this node is called the root node for that chunk
	//all replicated chunks timeout if it doesn't receive message from root
	
	//root chunks are all chunks that this node is responsible for replicating
	protected SortedSet<Long> rootChunks;	
	protected boolean treeInitialized;
	protected NetworkedManager networkedManager;
	protected TimerTask timeoutTask;
	protected TimerTask replicateTask;
	protected Timer timer;
	protected long nodeID;
	protected long rootLower;
	protected long rootUpper;
	public NetworkedReplicationManager(Manager m) {
		super(m);
		rootLower = 0;
		rootUpper = 0;
		nodeID = 0;
		treeInitialized = false;
		networkedManager = (NetworkedManager)manager;
		timeoutTask = new TimeoutTimer(this);
		replicateTask = new PushReplicateTimer(this);
		timer = new Timer(true);
	}
	
	public void run() {
		setDaemon(true);
	}
	
	public boolean isTreeInitialized() {
		return treeInitialized;
	}
	
	public void addIDToTree(ID id) {
		synchronized(getLock()) {
			long l = Long.parseLong(id.toString(10));
			if(l > rootLower && l < rootUpper) {
				rootChunks.add(l);
			}
		}
	}
	
	public void deleteIDFromTree(ID id) {
		synchronized(getLock()) {
			long l = Long.parseLong(id.toString(10));
			rootChunks.remove(l);
		}
	}
	
	//called by Manager 
	public void updateRootChunks(LeafSet l) {
		if(treeInitialized == false) {
			initializeRootChunks(l);
			//TODO: come up with a less arbitrary schedule
			//call timeout every day starting in one hour
			timer.schedule(timeoutTask, 1000*60*60, 1000*60*60*24);
			//call replicate every day starting in 30 mins
			timer.schedule(replicateTask, 1000*60*30, 1000*60*60*24);
		}
		else {
			update(l);
		}
	}
	
	protected void initializeRootChunks(LeafSet l) {
		synchronized(getLock()) {
			long lower = 0;		//the lowest chunkID that this node is respnsible for
			long upper = Long.MAX_VALUE;	//the highest id that this node is responsible for
			RoutingNode[] nodes = l.toArray();
			long last = 0;
			//nodes are in sorted order
			for(RoutingNode r : nodes) {
				long id = Long.parseLong(r.getId().toString(10));
				if(id < nodeID) {
					last = id;
				}
				else if(id > nodeID) {
					lower = (nodeID+last)/2;
					upper = (nodeID+id)/2;	
					break;
				}
			}
			//in future have storage presort the chunk ids
			rootChunks = storage.allChunks().subSet((new Long(lower)), (new Long(upper)));
			treeInitialized = true;
			rootLower = lower;
			rootUpper = upper;
		}
	}
	
	public void setCurrentNode(ID id) {
		nodeID = Long.parseLong(id.toString(10));
	}
	
	//TODO: only update new nodes rather than entire tree
	protected void update(LeafSet l) {
		synchronized(getLock()) {
			long lower = 0;		
			long upper = Long.MAX_VALUE;
			RoutingNode[] nodes = l.toArray();
			long last = 0;
			//nodes are in sorted order
			//must iterate through the nodes since we don't know what part of
			//the leafset changed
			for(RoutingNode r : nodes) {
				long id = Long.parseLong(r.getId().toString(10));
				if(id < nodeID) {
					last = id;
				}
				else if(id > nodeID) {
					lower = (nodeID+last)/2;
					upper = (nodeID+id)/2;	
					break;
				}
			}
			if(lower >= rootLower && upper <= rootUpper) {
				rootChunks = rootChunks.subSet((new Long(lower)), (new Long(upper)));
			}
			else {
				rootChunks = storage.allChunks().subSet((new Long(lower)), (new Long(upper)));
			}
			rootLower = lower;
			rootUpper = upper;
		}
	}
	
	public SortedSet<Long> getRootChunks() {
		return rootChunks;
	}
	
	public Object getLock() {
		return networkedManager.getLeafSetLock();
	}
	
	public Storage getStorage() {
		return storage;
	}
	
	//replicate chunks
	public void replicate(byte[] data, ID id, boolean firstReplication) {
		//chunkID is the id of the file that we want to replicate
		long chunkID = Long.parseLong(id.toString(10));
		ID[] ids = ((NetworkedManager)manager).getLeafSet().closest(id, 3);
		for(ID nodeID : ids) {
			//nid is the node that we want to send the chunk to
			long nid = Long.parseLong(nodeID.toString(10));
			ReplicationChunk repData = new ReplicationChunk(chunkID, data);
			byte[] repMessage = null;

			String type = "REPLICATE_PUT";
			//target node may already have this chunk so don't send data in order
			//to conserve bandwidth.
			if(firstReplication) {
				//send data to replicate
				type = "REPLICATE_INITIAL_PUT";
				try {
					repMessage = repData.toByteArray();
				}
				catch(Exception e) {
					System.err.println("could not convert chunk to byte[]");
					continue;
				}
			}
			
			Message m = new Message(nid, type, repMessage);
			manager.route(m);
		}
	}
	
	public void replicate(byte[] data, ID id) {
		replicate(data, id, false);
	}
	
	public void replicateDelete(ID id) {
		try {
			long chunkID = Long.parseLong(id.toString(10));
			ID[] ids = ((NetworkedManager)manager).getLeafSet().closest(id, 5);
			for(ID nodeID : ids) {
				long nid = Long.parseLong(nodeID.toString(10));
				ReplicationChunk repData = new ReplicationChunk(chunkID, null);
				Message m = new Message(nid, "REPLICATE_DELETE", repData.toByteArray());
				manager.route(m);
			}
		}
		catch(Exception e) {
			System.err.println("could not convert chunk to byte[]");
		}
	}
	
	protected class TimeoutTimer extends TimerTask {
		protected NetworkedReplicationManager nrm;
		public TimeoutTimer(NetworkedReplicationManager n) {
			super();
			nrm = n;
		}
		
		public void run() {
			synchronized(nrm.getLock()) {
				Storage s  = nrm.getStorage();
				TreeSet<Long> chunks = s.allChunks();
				SortedSet<Long> rootChunks = nrm.getRootChunks();
				long timeout = 1000*60*60*24*14;	//14 days
				for(long l : chunks) {
					if(!rootChunks.contains(l)) {
						s.deleteIfOld(new LongID(l), timeout);
					}
				}
			}
		}
	}
	
	protected class PushReplicateTimer extends TimerTask {
		protected NetworkedReplicationManager nrm;
		public PushReplicateTimer(NetworkedReplicationManager n) {
			super();
			nrm = n;
		}
		
		public void run() {
			synchronized(nrm.getLock()) {
				SortedSet<Long> chunkIDs = nrm.getRootChunks();
				for(long l : chunkIDs) {
					replicate(null, new LongID(l),false);
				}
			}
		}
	}
}
